package com.klun.concDesign.chapterFive.NIO;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MultiThreadServer {

    private Selector selector;
    private ExecutorService tp = Executors.newCachedThreadPool();

    // 用于时间统计
    public static Map<Socket,Long> time_stat = new HashMap<>(1024);

    public void startServer() throws IOException {
        // 通过工厂方法获取 selector 实例
        selector = SelectorProvider.provider().openSelector();
        // 获取服务端的 SocketChannel 实例
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false); // 设置 ssc 为非阻塞的模式
        InetAddress localHost = InetAddress.getLocalHost();
        System.out.println(localHost);
        // 进行端口绑定 此处写死
        InetSocketAddress isa = new InetSocketAddress(InetAddress.getLocalHost(), 8765);
        // InetSocketAddress isa = new InetSocketAddress(8765);
        ssc.socket().bind(isa);

        // 将 ServerSocketChannel 绑定到 Selector 上 并注册它 感兴趣的时间为 Accept
        // 这样这个 selector 就能为这个 channel 服务了
        // SelectionKey 表示一对 selector和channel 的关系 当selector 或者channel关闭时 SelectionKey 就会失效
        SelectionKey acceptKey = ssc.register(selector, SelectionKey.OP_ACCEPT);

        // 死循环 等待-分发 网络消息
        for(;;){
            selector.select();// 一个阻塞方法，如果都没有准备好则等待
            Set<SelectionKey> readyKeys = selector.selectedKeys(); // 获取准备好的selectionKey
            Iterator<SelectionKey> iterator = readyKeys.iterator();
            long e  =0;
            while (iterator.hasNext()){
                SelectionKey sk = iterator.next();
                iterator.remove(); //  从集合中移除 避免重复使用
                // 是否可以接收
                if(sk.isAcceptable()){
                    doAccept(sk);
                }
                // sk 有效 且可读
                else if(sk.isValid() && sk.isReadable()){
                    System.out.println("开始**读取客户端数据");
                    if(!time_stat.containsKey(((SocketChannel)sk.channel()).socket())){
                        time_stat.put(((SocketChannel)sk.channel()).socket(),System.currentTimeMillis());
                    }
                    doRead(sk);
                }
                // sk 有效 且可写
                else if(sk.isValid() && sk.isWritable()){
                    System.out.println("开始**返回客户端数据");
                    doWrite(sk);
                    e = System.currentTimeMillis();
                    Long b = time_stat.remove(((SocketChannel) sk.channel()).socket());
                    System.out.println("spend:"+(e-b)+"ms");
                }
            }
        }
    }
    private void doAccept(SelectionKey sk){
        ServerSocketChannel server = (ServerSocketChannel) sk.channel();
        SocketChannel clientChannel; // 表示和客户端的通信通道
        try{
            clientChannel = server.accept();
            clientChannel.configureBlocking(false);

            // Register this channel for reading
            SelectionKey clientKey = clientChannel.register(selector, SelectionKey.OP_READ);
            // 新建一个对象实例 一个EchoClient 代表一个客户端
            EchoClient echoClient = new EchoClient();
            clientKey.attach(echoClient);

            InetAddress inetAddress = clientChannel.socket().getInetAddress();
            SocketAddress localAddress = clientChannel.getLocalAddress();
            SocketAddress localAddress1 = clientChannel.getLocalAddress();
            System.out.println("Accepted connection from "+clientChannel.getRemoteAddress()+".");

        } catch (Exception e){
        }
    }

    private void doRead(SelectionKey sk){
        System.out.println("服务端：开始读 doRead() ");
        SocketChannel channel = (SocketChannel)sk.channel();
        ByteBuffer bb = ByteBuffer.allocate(20);
        int len;
        try{
            len = channel.read(bb);
            if(len<0){
                disconnect(sk);
                System.out.println("断开连接");
                return;
            }
        } catch (IOException e) {
            System.out.println("****** Failed to read from client");
            e.printStackTrace();
            disconnect(sk);
            return;
        }
        bb.flip();
        // 使用线程实例获取的消息
        tp.execute(new HandleMsg(sk,bb));
    }
    // 内部类
    public class HandleMsg implements Runnable {
        SelectionKey sk;
        ByteBuffer bb;

        public HandleMsg(SelectionKey sk, ByteBuffer bb) {
            this.sk = sk;
            this.bb = bb;
        }

        @Override
        public void run() {
            EchoClient echoClient = (EchoClient)sk.attachment();
            echoClient.enqueue(bb);
            // 注册感兴趣的 事件
            sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
            // 强迫 selector 立即返回
            selector.wakeup(); //???? 提交事件
        }
    }

    private void doWrite(SelectionKey sk){
        System.out.println("服务端：开始写 doWrite()");
        SocketChannel channel = (SocketChannel)sk.channel();
        EchoClient echoClient = (EchoClient)sk.attachment();
        LinkedList<ByteBuffer> outq = echoClient.getOutPutQueue();

        ByteBuffer bb = outq.getLast();
        //bb.put(new String("hello client!\r\n").getBytes());
        try {
            channel.read(bb);
        } catch (IOException e) {
            e.printStackTrace();
        }
        byte[] array = bb.array();
        String msg = new String(array).trim();
        System.out.println("----------------Msg"+msg);


        try{
            int len = channel.write(bb);

            if(len == -1){
                disconnect(sk);
                return;
            }

            if(bb.remaining() == 0){
                outq.removeLast();
            }
        } catch (IOException e) {
            System.out.println(" IOException Failed to write to client");
            e.printStackTrace();
            disconnect(sk);
        }

        if(outq.size() == 0){
            sk.interestOps(SelectionKey.OP_READ);
        }
    }
    private void disconnect(SelectionKey sk){
        System.out.println("&&&***** 断开连接");
        sk.cancel();
    }

}
